package com.atguigu.tingshu.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.LongTermsBucket;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.HitsMetadata;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.album.client.BaseCategoryFeignClient;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.BaseCategory3;
import com.atguigu.tingshu.model.album.BaseCategoryView;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.model.user.UserInfo;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.search.dao.AlbumInfoDao;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.user.client.UserInfoFeignClient;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import jakarta.annotation.Resource;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.DateTime;
import org.redisson.api.RBloomFilter;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class SearchServiceImpl implements SearchService {

  @Resource
  private BaseCategoryFeignClient baseCategoryFeignClient;

  @Resource
  private ElasticsearchClient elasticsearchClient;

  @Autowired
  private AlbumInfoFeignClient albumInfoFeignClient;

  @Autowired
  private UserInfoFeignClient userInfoFeignClient;

  @Autowired
  private ThreadPoolExecutor threadPoolExecutor;

  @Autowired
  private RBloomFilter<Object> rBloomFilter;

  @Autowired
  private RedisTemplate redisTemplate;

  @Autowired
  private AlbumInfoDao albumInfoDao;
  /**
   * 查询排行榜数据: 前10
   * @param category1Id
   * @param type
   * @return
   */
  @Override
  public Object findRankingList(Long category1Id, String type) {
    //获取日期
    String time = new DateTime().toString("yyyyMMdd");
    //获取指定一级分类的指定类型的排行榜前10的数据
    Set<Long> albumIds =
            redisTemplate.opsForZSet().range(
                    category1Id + ":" + type + ":"+ time,
                    0,
                    10);
    //查询这些数据的详细专辑的数据
    Iterator<AlbumInfoIndex> iterator =
            albumInfoDao.findAllById(albumIds).iterator();
    List<AlbumInfoIndex> albumInfoIndexList = new ArrayList<>();
    //遍历
    while (iterator.hasNext()){
      albumInfoIndexList.add(iterator.next());
    }
    return albumInfoIndexList;
  }


  /**
   * 1108 查询专辑详细页面信息 异步编排优化
   * @param albumId
   * @return Object
   */
  @Override
  public Object getAlbumInfo(Long albumId) {
    // 1108 校验布隆过滤器
    if(!rBloomFilter.contains("albumId:" + albumId)){
      throw new GuiguException(201, "专辑不存在!");
    }
    // 初始化返回的数据对象 线程安全
    Map<String, Object> result = new ConcurrentHashMap<>();

    // 任务一: 获取专辑信息
    CompletableFuture<AlbumInfo> future1 = CompletableFuture.supplyAsync(() -> {
      // 查询专辑的详细信息
      AlbumInfo albumInfo = albumInfoFeignClient.getAlbumInfo(albumId);
      // 非空判断
      if (albumInfo == null || albumInfo.getId() == null) {
        return null;
      }
      // 封装数据
      result.put("albumInfo", albumInfo);
      // 返回
      return albumInfo;
    }, threadPoolExecutor);

    // 任务二: 获取专辑统计信息
    CompletableFuture<Void> future2 = future1.thenAcceptAsync(albumInfo -> {
      // 非空判断
      if (albumInfo == null) {
        return;
      }
      // 查询专辑的统计数据
      Map<String, String> albumStatMap = albumInfoFeignClient.getAlbumStat(albumId);
      AlbumStatVo albumStatVo = new AlbumStatVo();
      albumStatVo.setPlayStatNum(Integer.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_PLAY))); // 播放量
      albumStatVo.setSubscribeStatNum(Integer.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_SUBSCRIBE))); // 订阅量
      albumStatVo.setBuyStatNum(Integer.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_BROWSE))); // 购买量
      albumStatVo.setCommentStatNum(Integer.valueOf(albumStatMap.get(SystemConstant.ALBUM_STAT_COMMENT))); // 评论量
      result.put("albumStatVo", albumStatVo);
    }, threadPoolExecutor);

    // 任务三: 获取分类信息
    CompletableFuture<Void> future3 = future1.thenAcceptAsync(albumInfo -> {
      // 非空判断
      if (albumInfo == null) {
        return;
      }
      // 查询专辑的分类数据
      BaseCategoryView baseCategoryView = baseCategoryFeignClient.getBaseCategoryView(albumInfo.getCategory3Id());
      result.put("baseCategoryView", baseCategoryView);
    }, threadPoolExecutor);

    // 任务四: 获取主播信息
    CompletableFuture<Void> future4 = future1.thenAcceptAsync(albumInfo -> {
      // 非空判断
      if (albumInfo == null) {
        return;
      }
      // 查询专辑的作者数据
      UserInfo userInfo = userInfoFeignClient.getUserInfo(albumInfo.getUserId());
      UserInfoVo userInfoVo = new UserInfoVo();
      BeanUtils.copyProperties(userInfo, userInfoVo);
      result.put("announcer", userInfoVo);
    }, threadPoolExecutor);

    // 全部任务结束
    CompletableFuture.allOf(future2, future3, future4).join();
    // 返回
    return result;
  }


  /**
   * 1106 首页根据一级分类id查询一级分类id下前7热门三级分类下前6热门专辑数据
   * @param category1Id
   * @return
   */
  @SneakyThrows
  @Override
  public Object channel(Long category1Id) {
    // 根据一级分类id查询前7热门的三级分类的信息
    List<BaseCategory3> baseCategory3List
            = baseCategoryFeignClient.getBaseCategory3List(category1Id);
    // 将list转化为map key=三级分类的id value=三级分类数据项
    Map<Long, BaseCategory3> category3Map = baseCategory3List.stream().collect(Collectors.toMap(
            key -> key.getId(),
            val -> val
    ));
    // 将每个三级分类的id转化为es搜索的条件类型
    List<FieldValue> fieldValueList = baseCategory3List.stream().map(
            baseCategory3 -> FieldValue.of(baseCategory3.getId())).toList();
    // es搜索的请求对象初始化
    SearchRequest.Builder builder = new SearchRequest.Builder();
    // 指定索引: 不指定索引会报错
    builder.index("albuminfo_java0523");

    // 将7个三级分类id(不分词)当做查询条件去es查询这7个三级分类的全部的专辑数据
    builder.query( // 构建查询条件
            query -> query.terms( // 使用词条查询不分词
                  terms -> terms.field("category3Id").terms( // 指定三级分类id域作为查询条件
                    termsValue -> termsValue.value(fieldValueList) // 指定三级分类的id作为参数
                  )
            )
    ); // 相当于where

    // 聚合条件: 根据三级分类id进行分桶, 还要对桶里面的数据进行hotScore的降序 限制前6
    builder.aggregations(
            "aggCategory3AlbumList", // 桶的名字
            agg -> agg.terms(
                    terms -> terms.field("category3Id")
            ).aggregations(
                    "aggHotScore", // 排序桶名
                        subAgg -> subAgg.topHits(
                            topHits -> topHits.sort(
                                    sort -> sort.field(
                                            f -> f.field("hotScore").order(SortOrder.Desc)) // 根据热度倒序排名
                            ).size(6) // 前六的数据
                        )
            )
    ); // 相当于group by

    // 执行搜索
    SearchResponse<AlbumInfoIndex> searchResponse
            = elasticsearchClient.search(builder.build(), AlbumInfoIndex.class);

    // 解析结果集
    // List<AlbumInfoIndex> albumInfoIndexList =
    //         searchResponse.aggregations().get("aggCategory3AlbumList").topHits().hits().hits().stream().map(
    //           h -> h.source().to(AlbumInfoIndex.class)
    // ).toList();

    // 获取到七个桶的集合
    List<LongTermsBucket> longTermsBucketList
            = searchResponse.aggregations().get("aggCategory3AlbumList").lterms().buckets().array();
    // 遍历集合
    List<JSONObject> jsonObjectList = longTermsBucketList
            .stream().map(
                    bucket -> {
                      // 获取到一个桶的数据
                      List<AlbumInfoIndex> albumInfoIndexList =
                              bucket.aggregations().get("aggHotScore").topHits().hits().hits().stream()
                                    .map(score -> score.source().to(AlbumInfoIndex.class))
                                    .toList();
                      // Long category3Id = albumInfoIndexList.get(0).getCategory3Id();
                      // System.out.println(category3Id + "------三级分类id---------" + albumInfoIndexList.toString());
                      // 保存专辑的数据
                      JSONObject result = new JSONObject();
                      result.put("list", albumInfoIndexList);
                      // 获取三级分类的id 根据三级分类id获取三级分类对象
                      Long category3Id = albumInfoIndexList.get(0).getCategory3Id();
                      // 保存三级分类的数据
                      result.put("baseCategory3", category3Map.get(category3Id));
                      // 返回
                      return result;
                    }).toList();

    // 返回
    return jsonObjectList;
  }

  /**
   * 1107 专辑搜索
   * @param albumIndexQuery
   * @return Result
   */
  @SneakyThrows
  @Override
  public Object search(AlbumIndexQuery albumIndexQuery) {
    // 构建查询条件
    SearchRequest searchRequest = buildQueryParams(albumIndexQuery);
    // 执行搜索
    SearchResponse<AlbumInfoIndex> searchResponse
            = elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);
    // 解析结果
    AlbumSearchResponseVo result = getSearchResult(searchResponse);
    result.setPageNo(albumIndexQuery.getPageNo());
    // 获取每页显示的条数
    Integer pageSize = albumIndexQuery.getPageSize();
    result.setPageSize(pageSize);
    // 获取总数据量
    Long total = result.getTotal();
    // 计算总页数
    result.setTotalPages(total % pageSize == 0 ? total/pageSize : total/pageSize + 1);
    // 返回
    return result;
  }



  /**
   * 构建查询条件
   * @param albumIndexQuery
   * @return SearchRequest
   */
  private SearchRequest buildQueryParams(AlbumIndexQuery albumIndexQuery) {
    // 条件构造器初始化
    SearchRequest.Builder builder = new SearchRequest.Builder();
    // 指定索引
    builder.index("albuminfo_java0523");
    // 构建组合查询条件
    BoolQuery.Builder boolQueryBuilder = new BoolQuery.Builder();
    // 第一层筛选 关键词条件
    String keyword = albumIndexQuery.getKeyword();
    if (StringUtils.isNotEmpty(keyword)){
      // 和专辑的名字匹配
      boolQueryBuilder.should(
              s -> s.match(
                      m -> m.field("albumTitle").query(keyword)
              )
      );
      // 和专辑的简介匹配
      boolQueryBuilder.should(
              s -> s.match(
                      m -> m.field("albumIntro").query(keyword)
              )
      );
      // 和专辑的作者名字匹配
      boolQueryBuilder.should(
              s -> s.match(
                      m -> m.field("announcerName").query(keyword)
              )
      );
    }
    // 第二层筛选 分类条件
    // 一层分类
    Long category1Id = albumIndexQuery.getCategory1Id();
    if (category1Id != null){
      boolQueryBuilder.filter(
              f -> f.term(t -> t.field("category1Id").value(category1Id))
      );
    }
    // 二层分类
    Long category2Id = albumIndexQuery.getCategory2Id(); // 前端没传
    if (category2Id != null){
      boolQueryBuilder.filter(
              f -> f.term(t -> t.field("category2Id").value(category2Id))
      );
    }
    // 三层分类
    Long category3Id = albumIndexQuery.getCategory3Id();
    if (category3Id != null){
      boolQueryBuilder.filter(
              f -> f.term(t -> t.field("category3Id").value(category3Id))
      );
    }
    // 第三层筛选 标签条件查询
    List<String> attributeList = albumIndexQuery.getAttributeList();
    if (attributeList != null && !attributeList.isEmpty()){
      // 将每个标签条件都进行构建
      attributeList.stream().forEach(attrString -> {
        // 切分
        String[] split = attrString.split(":");

        boolQueryBuilder.filter(f -> f.nested(
                nested ->
                        nested.path("attributeValueIndexList")
                              .query(
                                q -> q.bool(
                                  b -> b.must(m -> m.term(t -> t.field("attributeValueIndexList.attributeId").value(split[0])))
                                        .must(m -> m.term(t -> t.field("attributeValueIndexList.valueId").value(split[1])))
                                )
                              )
                    )
            );
        }
      );

    }
    // 保存查询条件
    builder.query(boolQueryBuilder.build()._toQuery());
    // 排序 1: 综合排序 2: 播放量 3: 最近更新
    String order = albumIndexQuery.getOrder(); // 3:desc
    if (StringUtils.isNotEmpty(order)){
      String[] split = order.split(":");
      // 判断是正序还是倒序
      SortOrder sortOrder = split[1].equals("desc") ? SortOrder.Desc : SortOrder.Asc;
      switch (split[0]){
        // 综合排序
        case "1" -> builder.sort(
                s -> s.field(
                        f -> f.field("hotScore").order(sortOrder)
                )
        ); // 包含隐式的break
        // 播放量排序
        case "2" -> builder.sort(
                s -> s.field(
                        f -> f.field("playStatNum").order(sortOrder)
                )
        );
        // 时间排序
        case "3" -> builder.sort(
                s -> s.field(
                        f -> f.field("createTime").order(sortOrder)
                )
        );
      }
    } else {
      // 默认排序
      builder.sort(s -> s.field(
              f -> f.field("id").order(SortOrder.Desc)
      ));
    }
    // 分页
    Integer pageNo = albumIndexQuery.getPageNo();
    Integer pageSize = albumIndexQuery.getPageSize();
    builder.from((pageNo - 1) * pageSize); // 起始偏移量
    builder.size(pageSize); // 每页数量
    // 高亮
    builder.highlight(
            h -> h.fields(
                    "albumTitle",
                    high -> high.preTags("<font style=color:green>").postTags("</font>")
            )
    );
    // 返回
    return builder.build();
  }

  /**
   * 解析查询条件
   *
   * @param searchResponse
   * @return null
   */
  private AlbumSearchResponseVo getSearchResult(SearchResponse<AlbumInfoIndex> searchResponse) {
    // 创建要返回的包装结果
    AlbumSearchResponseVo albumSearchResponseVo = new AlbumSearchResponseVo();
    // 获取命中的数据对象
    HitsMetadata<AlbumInfoIndex> hits = searchResponse.hits();
    // 获取总命中的数据量
    long value = hits.total().value();
    albumSearchResponseVo.setTotal(value);
    // 获取命中的详细数据
    List<AlbumInfoIndexVo> list = hits.hits().stream().map(
            h -> {
              // 获取原始数据
              AlbumInfoIndex albumInfoIndex = h.source();
              // 返回结果初始化
              AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
              // 属性转移
              BeanUtils.copyProperties(albumInfoIndex, albumInfoIndexVo);
              // 获取高亮的数据
              List<String> highlightList = h.highlight().get("albumTitle");
              if (highlightList != null && !highlightList.isEmpty()){
                // 将集合元素拼接成字符串
                String title = String.join("", highlightList);
                // String title = highlightList.stream().collect(Collectors.joining());
                // 覆盖原数据
                albumInfoIndexVo.setAlbumTitle(title);
              }
              return albumInfoIndexVo;
            }
    ).toList();
    // 记录返回结果
    albumSearchResponseVo.setList(list);
    // 返回
    return albumSearchResponseVo;
  }

  /**
   * 1107 提示词功能
   * @param keywords
   * @return Result
   */
  @SneakyThrows
  @Override
  public Object completeSuggest(String keywords) {
    // 条件构造器初始化
    SearchRequest.Builder builder = new SearchRequest.Builder();
    // 设置索引
    builder.index("suggestinfo_java0523");
    // 拼接条件
    builder.suggest(
            s -> s.suggesters(
                    "suggestKeywords",
                    suggest -> suggest.prefix(keywords).completion(
                            c -> c.field("keyword")
                                  .fuzzy(f -> f.fuzziness("auto")) // 偏移量使用默认
                                  .skipDuplicates(true) // 去重
                    )
            ).suggesters(
                    "suggestKeywordsPinyin",
                    suggest -> suggest.prefix(keywords).completion(
                            c -> c.field("keywordPinyin")
                                  .fuzzy(f -> f.fuzziness("auto")) // 偏移量使用默认
                                  .skipDuplicates(true) // 去重
                    )
            ).suggesters(
                    "suggestKeywordSequence",
                    suggest -> suggest.prefix(keywords).completion(
                            c -> c.field("keywordSequence")
                                  .fuzzy(f -> f.fuzziness("auto")) // 偏移量使用默认
                                  .skipDuplicates(true) // 去重
                    )
            )
    );
    // 执行搜索
    SearchResponse<SuggestIndex> searchResponse
            = elasticsearchClient.search(builder.build(), SuggestIndex.class);
    // 解析结果
    Map<String, List<Suggestion<SuggestIndex>>> suggestMap
            = searchResponse.suggest();
    // 解析关键字结果
    List<String> list1 = getSuggestResult(suggestMap, "suggestKeywords");
    // 解析拼音结果
    List<String> list2 = getSuggestResult(suggestMap, "suggestKeywordsPinyin");
    // 解析首字母结果
    List<String> list3 = getSuggestResult(suggestMap, "suggestKeywordSequence");
    // 三个list合并为一个list并且去重
    List<String> list = Stream.of(list1, list2, list3) // 合并三个集合为一个数据流
                              .flatMap(Collection::stream) // 使用集合工具类进行流式操作
                              .collect(Collectors.toSet()) // 转换为set集合去重
                              .stream().limit(10) // 限制最多10条数据
                              .toList();
    return list;
  }


  /**
   * 解析提示词的结果
   *
   * @param suggestMap
   * @return
   */
  private List<String> getSuggestResult(Map<String, List<Suggestion<SuggestIndex>>> suggestMap, String suggestKeywords) {
    // 返回结果初始化
    List<String> result = new ArrayList<>();
    // 获取指定别名的提示词的搜索结果
    List<Suggestion<SuggestIndex>> list = suggestMap.get(suggestKeywords);
    // 非空判断
    if (list != null && !list.isEmpty()){
      // 获取匹配到的包装数据
      Suggestion<SuggestIndex> indexSuggestion = list.get(0);
      // 获取最终的结果
      result = indexSuggestion.completion().options().stream().map(
              s -> s.source().getTitle()
      ).toList();
    }
    return result;
  }



}
